package com.trip.order.producer;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.trip.common.util.ListSplitter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.Collectors;

@Component
@Slf4j
public class OrderProducer extends AbstractRocketMQService{
    @Autowired
    private RocketMQTemplate rocketMqTemplate;

    /**
     * 原生的producer
     */
    @Autowired
    private DefaultMQProducer producer;

    @Override
    public boolean sendMessage(String destination, String msg) {
        String[] split = destination.split(":");
        if (split.length == 2) {
            return this.sendMessage(split[0], split[1], msg);
        }
        return this.sendMessage(destination, null, msg);
    }

    @Override
    public boolean sendMessage(String topicName, String tags, String msg) {
        return this.sendMessage(topicName, tags, null, msg);
    }

    @Override
    public boolean sendMessage(String topicName, String tags, String key, String msg) {
            MessageBuilder<?> messageBuilder = MessageBuilder.withPayload(msg);
            //设置key,唯一标识码要设置到keys字段，方便将来定位消息丢失问题
            if (StrUtil.isNotBlank(key)) {
                messageBuilder.setHeader(MessageConst.PROPERTY_KEYS, key);
            }
            Message<?> message = messageBuilder.build();
            SendResult sendResult = rocketMqTemplate.syncSend(StrUtil.isBlank(tags) ? topicName : (topicName + ":" + tags), message);
            if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
                log.info("MQ发送同步消息成功,topicName={},tags={},msg={},sendResult={}", topicName, tags, msg, sendResult);
                return true;
            } else {
                log.warn("MQ发送同步消息不一定成功,topicName={},tags={},msg={},sendResult={}", topicName, tags, msg, sendResult);
                return false;
            }
    }

    @Override
    public void sendAsyncMessage(String destination, Object msg, SendCallback sendCallback) {
        this.rocketMqTemplate.asyncSend(destination,
                msg instanceof String ? msg : JSON.toJSONString(msg),
                sendCallback);
        log.info("MQ发送异步消息,destination={} msg={}", destination, msg);
    }

    @Override
    public void sendBatchMessage(String topicName, String tags, Long timeout, List<?> list) {
        //转为message
        List<org.apache.rocketmq.common.message.Message> messages = list.stream().map(x ->
                new org.apache.rocketmq.common.message.Message(topicName, tags,
                        //String类型不需要转JSON，其它类型都要转为JSON模式
                        x instanceof String ? ((String) x).getBytes(StandardCharsets.UTF_8) : JSON.toJSONBytes(x))
        ).collect(Collectors.toList());

        //自动分割发送，把大的消息分裂成若干个小的消息(每次发送最大只能4MB)
        ListSplitter splitter = new ListSplitter(messages);

        while (splitter.hasNext()) {
            try {
                List<org.apache.rocketmq.common.message.Message> listItem = splitter.next();
                SendResult sendResult = producer.send(listItem, timeout == null ? 30000L : timeout);
                if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
                    log.info("MQ发送批量消息成功,topicName={}  tags={}， size={},sendResult={}", topicName, tags, listItem.size(), sendResult);
                } else {
                    log.warn("MQ发送批量消息不一定成功,topicName={}  tags={}， size={},sendResult={}", topicName, tags, listItem.size(), sendResult);
                }
            } catch (Exception e) {
                //处理error
                log.error("MQ发送批量消息失败,topicName={}  tags={}，,errorMessage={}", topicName, tags, e.getMessage(), e);
                throw new RuntimeException("MQ发送批量消息失败,原因：" + e.getMessage());
            }
        }
    }
}
